package com.jd.platform.gobrs.worker.keylistener;

import cn.hutool.core.date.SystemClock;
import com.github.benmanes.caffeine.cache.Cache;
import com.jd.platform.gobrs.common.model.GCacheKeyModel;
import com.jd.platform.gobrs.common.model.GCacheKeyModel;
import com.jd.platform.gobrs.common.model.KeyTraffic;
import com.jd.platform.gobrs.common.rule.KeyRule;
import com.jd.platform.gobrs.worker.cache.CaffeineCacheHolder;
import com.jd.platform.gobrs.worker.netty.pusher.DashboardPusher;
import com.jd.platform.gobrs.worker.netty.pusher.IPusher;
import com.jd.platform.gobrs.worker.rule.KeyRuleHolder;
import com.jd.platform.gobrs.worker.starters.EtcdStarter;
import com.jd.platform.gobrs.worker.tool.SlidingWindow;
import org.checkerframework.checker.units.qual.K;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.LongAdder;

/**
 * @program: gobrs
 * @ClassName AutoKeyListener
 * @description:
 * @author: sizegang
 * @create: 2022-01-09 22:45
 * @Version 1.0
 **/
@Component
public class AutoKeyListener extends AbstractAutoKeyListener {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private DashboardPusher dashboardPusher;

    @Resource
    ManualKeyListener manualKeyListener;

    @Resource
    private ThreadPoolTaskExecutor executeHitSumThreadPool;


    public static Map<String, Map<String, KeyTraffic>> calculateBuffer = new ConcurrentHashMap<>(); // 计算hit 缓冲

    @Override
    public void doNewKey(GCacheKeyModel gCacheKeyModel, KeyEventOriginal orignal, String key) {
        // 获取规则 如果设置了规则 则走规则逻辑
        KeyRule ruleByAppAndKey = KeyRuleHolder.getRuleByAppAndKeyforAuto(gCacheKeyModel);
        // 兼容 manual模式
        if (ruleByAppAndKey != null) { // 如果有规则 则说明已经被设置了具体规则了
            manualKeyListener.doNewKey(gCacheKeyModel, orignal, key);
            // 有规则的就不统计了 只监控没有走缓存的数据
            return;
        }
        KeyTraffic keyTraffic = buildKeyTraffic(gCacheKeyModel);

        dashboardPusher.pushAutoKey(sumPushedHit(keyTraffic));
    }

    @Override
    public void removeKey(GCacheKeyModel hotKeyModel, KeyEventOriginal orignal) {
    }


    /**
     * 计算统计 考虑问题 线程安全 性能  在这里 性能优先级要高于安全 统计的次数可以不是100%准确 但是 高并发下的计算性能一定要保证 队列不卡住
     */
    private KeyTraffic sumPushedHit(KeyTraffic newKeyTraffic) {
        Future<?> submit = executeHitSumThreadPool.submit(() -> {
            Map<String, KeyTraffic> appCacheMap = calculateBuffer.get(newKeyTraffic.getApp());
            String key = newKeyTraffic.getKey();
            return noSafeSum(newKeyTraffic, appCacheMap, key);
        });
        try {
            return (KeyTraffic) submit.get();
        } catch (Exception e) {
            e.printStackTrace();
            return newKeyTraffic;
        }
    }

    private KeyTraffic noSafeSum(KeyTraffic newKeyTraffic, Map<String, KeyTraffic> appCacheMap, String key) {
        if (appCacheMap != null) {
            KeyTraffic keyTraffic = appCacheMap.get(key);
            if (keyTraffic == null) {
                appCacheMap.put(key, newKeyTraffic);
                return newKeyTraffic;
            } else {
                keyTraffic.getHit().add(1);
                newKeyTraffic.setHits(keyTraffic.getHit().sum());
                return newKeyTraffic;
            }
        } else {
            calculateBuffer.put(newKeyTraffic.getApp(), new ConcurrentHashMap<String, KeyTraffic>());
        }
        return newKeyTraffic;
    }

    // 不实用beanutils 为什么 为了快 为什么不直接用 GCacheKeyModel 对象 也是为了快 越小的对象 越少的操作
    private KeyTraffic buildKeyTraffic(GCacheKeyModel gCacheKeyModel) {
        KeyTraffic keyTraffic = new KeyTraffic();
        keyTraffic.setKey(gCacheKeyModel.getKey());
        keyTraffic.setSize(gCacheKeyModel.getSize());
        keyTraffic.setHits(gCacheKeyModel.getCount());
        keyTraffic.setApp(gCacheKeyModel.getAppName());
        keyTraffic.setId(gCacheKeyModel.getId());
        keyTraffic.setTime(gCacheKeyModel.getCreateTime());
        return keyTraffic;
    }

    // synchronized jdk优化 这里就不用lock了
//    private synchronized void safeSum(GCacheKeyModel gCacheKeyModel, Map<String, GCacheKeyModel> appCacheMap, String key) {
//        if (appCacheMap != null) {
//            GCacheKeyModel GCacheKeyModel = appCacheMap.get(key);
//            if (GCacheKeyModel == null) {
//                appCacheMap.put(key, new GCacheKeyModel(gCacheKeyModel.getSize(), new LongAdder()));
//            } else {
//                GCacheKeyModel.getCount().add(gCacheKeyModel.getCount());// 计数
//            }
//            calculateBuffer.put(gCacheKeyModel.getAppName(), appCacheMap);
//        } else {
//            calculateBuffer.put(gCacheKeyModel.getAppName(), new ConcurrentHashMap<>());
//        }
//    }

//    public Map<String, Map<String, GCacheKeyModel>> autoKeyCalculateBuffer() {
//        return calculateBuffer;
//    }
}
